ποΈGitΠ―ΡΠ°ποΈ
Commit 1dac0f3ac9766d1b756b1b255e14207d1d73eb24
Parents : 7710220
Author : James Rich <2199651+jamesarich@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-08-05T19:57:35-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-08-06T00:57:35Z
fix(service): unwedge the inbound pipeline behind stale-Connected zombies (#6587)
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Changes
3 files changed, 190 insertions(+), 4 deletions(-)
Diff
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
index 3bb66621fb..709199b5a1 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
@@ -33,6 +33,7 @@ import org.meshtastic.core.common.di.ServiceScope
import org.meshtastic.core.common.util.handledLaunch
import org.meshtastic.core.common.util.nowMillis
import org.meshtastic.core.common.util.nowSeconds
+import org.meshtastic.core.common.util.safeCatching
import org.meshtastic.core.common.util.safeCatchingAll
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.DeviceType
@@ -136,7 +137,14 @@ class MeshConnectionManagerImpl(
// Bridge transport-level state into the canonical app-level state.
// This is the ONLY consumer of RadioInterfaceService.connectionState β it applies
// light-sleep policy and handshake awareness before writing to ServiceRepository.
- radioInterfaceService.connectionState.onEach(::onRadioConnectionState).launchIn(scope)
+ // Guarded per-emission: one uncaught throw here would kill the sole bridge collector and
+ // permanently freeze the app-level state (a stuck-"Connected" UI no transport event can fix).
+ radioInterfaceService.connectionState
+ .onEach { state ->
+ safeCatching { onRadioConnectionState(state) }
+ .onFailure { Logger.e(it) { "Connection state bridge failed for $state; collector kept alive" } }
+ }
+ .launchIn(scope)
// Ensure notification title and content stay in sync with state changes
serviceRepository.connectionState.onEach { updateStatusNotification() }.launchIn(scope)
diff --git a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
index 2a151d8aea..b52cf99199 100644
--- a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
+++ b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
@@ -27,10 +27,13 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
@@ -50,6 +53,8 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
import okio.ByteString.Companion.toByteString
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
@@ -271,7 +276,23 @@ class SharedRadioInterfaceService(
}
override suspend fun runWhileSessionActive(session: RadioSessionContext, block: suspend () -> Unit): Boolean =
- sessionOperationMutex.withLock { runWithSessionLease(session) { block() } }
+ sessionOperationMutex.withLock {
+ runWithSessionLease(session) {
+ // Bound the handler: it holds sessionOperationMutex (the whole inbound pipeline) and an admitted
+ // lease (which teardown's drain awaits), so an indefinite suspension here is a total wedge, not a
+ // slow packet. Cancelling the block releases both. Only OUR timeout is swallowed β ensureActive()
+ // rethrows if the surrounding scope was cancelled concurrently.
+ try {
+ withTimeout(SESSION_HANDLER_TIMEOUT_MILLIS) { block() }
+ } catch (timeout: TimeoutCancellationException) {
+ currentCoroutineContext().ensureActive()
+ Logger.e(timeout) {
+ "Session handler exceeded ${SESSION_HANDLER_TIMEOUT_MILLIS}ms and was cancelled; " +
+ "dropping its packet to keep the receive pipeline alive"
+ }
+ }
+ }
+ }
/** Runs a callback only while [session] still owns admission, atomically with session teardown. */
private inline fun runIfTransportSessionActive(session: RadioTransportSession, block: () -> Unit): Boolean =
@@ -299,7 +320,21 @@ class SharedRadioInterfaceService(
sessionDrainWaiter ?: CompletableDeferred<Unit>().also { sessionDrainWaiter = it }
}
}
- drainWaiter?.await()
+ if (drainWaiter != null) {
+ // The drain must complete before a replacement generation is admitted (DB-atomicity contract), so we
+ // keep waiting β but never silently. A lease stuck past the handler timeout means a handler ignored
+ // cancellation; these error-level reports are the observability surface for that wedge (this wait
+ // previously blocked disconnect()/restart forever with no telemetry at all).
+ var waitedMillis = 0L
+ while (withTimeoutOrNull(DRAIN_WAIT_LOG_INTERVAL_MILLIS) { drainWaiter.await() } == null) {
+ waitedMillis += DRAIN_WAIT_LOG_INTERVAL_MILLIS
+ val outstanding = synchronized(sessionCallbackLock) { admittedSessionOperations }
+ Logger.e {
+ "Transport teardown blocked ${waitedMillis}ms waiting for $outstanding admitted session " +
+ "operation(s) to release (generation=${session.generation})"
+ }
+ }
+ }
synchronized(sessionCallbackLock) {
if (activeTransportSession === session) {
check(admittedSessionOperations == 0) { "Session revoked before admitted operations drained" }
@@ -432,6 +467,18 @@ class SharedRadioInterfaceService(
* flaky GATT connection. Serial and TCP typically flush well under this window.
*/
private const val POLITE_DISCONNECT_DRAIN_MS = 500L
+
+ /**
+ * Ceiling on a single [runWhileSessionActive] handler. The block holds [sessionOperationMutex] β the whole
+ * inbound pipeline β so a handler that suspends indefinitely wedges packet processing, fills [_receivedData],
+ * and deadlocks teardown's lease drain (a Connected-looking zombie only a force-stop clears; observed in the
+ * field as multi-hour "receive queue at capacity" sessions). 2 minutes is far above any legitimate handler
+ * (large-mesh config DB installs run seconds) while still bounding the wedge.
+ */
+ private const val SESSION_HANDLER_TIMEOUT_MILLIS = 2 * 60 * 1000L
+
+ /** How often [revokeTransportSession] reports a lease drain that has not completed. */
+ private const val DRAIN_WAIT_LOG_INTERVAL_MILLIS = 15 * 1000L
}
private val initLock = Mutex()
@@ -979,7 +1026,6 @@ class SharedRadioInterfaceService(
@Suppress("TooGenericExceptionCaught")
private fun enqueueReceivedData(bytes: ByteArray, session: RadioTransportSession) {
try {
- lastDataReceivedMillis = now()
// trySend synchronously onto the Channel so packet order matches arrival order. The
// previous `launch { emit() }` pattern dispatched each packet onto a fresh coroutine,
// letting the scheduler reorder them β which broke the firmware config handshake
@@ -993,6 +1039,12 @@ class SharedRadioInterfaceService(
}
val frame = ReceivedRadioFrame(payload = bytes.toByteString(), session = session.context)
val result = _receivedData.trySend(frame)
+ if (result.isSuccess) {
+ // Stamp liveness only for frames actually admitted to the queue. Stamping on arrival kept
+ // checkLiveness() satisfied while a wedged consumer dropped every frame β a Connected-looking zombie
+ // the watchdog existed to catch. A full queue now reads as silence and trips liveness recovery.
+ lastDataReceivedMillis = now()
+ }
if (result.isFailure) {
// Rate-limited on purpose: drops only happen under sustained inbound traffic, and Kermit forwards to
// Datadog/Crashlytics, so logging every drop would turn a bounded memory problem into unbounded
diff --git a/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt b/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt
index ea7a762857..5f5852e3f7 100644
--- a/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt
+++ b/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt
@@ -820,6 +820,132 @@ class SharedRadioInterfaceServiceLivenessTest {
}
}
+ /**
+ * Regression for the field wedge behind "app shows Connected but the node stops updating": frames dropped by a full
+ * receive queue must NOT feed the liveness timer. Before the fix, [SharedRadioInterfaceService] stamped
+ * `lastDataReceivedMillis` on arrival (even for dropped frames), so a wedged consumer kept liveness satisfied
+ * forever while every frame was discarded.
+ */
+ @Test
+ fun `frames dropped by a full receive queue do not reset the liveness timer`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+ try {
+ // No collector attached: fill the channel to capacity at t=0. All of these are admitted
+ // and stamp liveness at 0.
+ val payload = byteArrayOf(1)
+ repeat(SharedRadioInterfaceService.RECEIVE_QUEUE_CAPACITY) { service.handleFromRadio(payload) }
+
+ // A frame arriving at t=30s is DROPPED (queue full). It must not count as liveness data.
+ clock = 30_000L
+ service.handleFromRadio(payload)
+
+ // At t=65s the silence is 65s if the drop was correctly ignored (fires), but only 35s if
+ // the drop stamped the timer (must not happen).
+ clock = 65_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ assertTrue(
+ createdTransports.first().closeCalled,
+ "Liveness must fire on queue-full silence β dropped frames must not feed the timer",
+ )
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ // βββ Session handler timeout: the pipeline must not wedge forever βββββββββββββββββββββββββββ
+
+ /**
+ * Regression for the 2.8.0 stale-connection wedge: a handler that suspends indefinitely inside
+ * [SharedRadioInterfaceService.runWhileSessionActive] holds the session-operation lane (the whole inbound
+ * pipeline). It must be cancelled at the handler timeout so queued work behind it can run.
+ */
+ @Test
+ fun `wedged session handler is cancelled at the timeout and the pipeline continues`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+ val session = requireNotNull(service.activeSession.value)
+ val wedgeStarted = CompletableDeferred<Unit>()
+ val neverReleased = CompletableDeferred<Unit>()
+ var wedgeRanToCompletion = false
+
+ val wedged = launch {
+ service.runWhileSessionActive(session) {
+ wedgeStarted.complete(Unit)
+ neverReleased.await() // simulates a handler stuck on an unbounded suspension
+ wedgeRanToCompletion = true
+ }
+ }
+ wedgeStarted.await()
+ val nextStarted = CompletableDeferred<Unit>()
+ val next = launch { service.runWhileSessionActive(session) { nextStarted.complete(Unit) } }
+ try {
+ testDispatcher.scheduler.runCurrent()
+ assertFalse(nextStarted.isCompleted, "ordered work is serialized behind the wedged handler")
+
+ // Cross the 2-minute handler timeout: the wedged block is cancelled, releasing the lane.
+ advanceTimeBy(121_000L)
+ wedged.join()
+ next.join()
+
+ assertFalse(wedgeRanToCompletion, "the wedged handler must have been cancelled, not completed")
+ assertTrue(nextStarted.isCompleted, "the handler timeout must release the pipeline for queued work")
+ } finally {
+ neverReleased.complete(Unit)
+ wedged.cancel()
+ next.cancel()
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ /**
+ * The user-facing half of the same wedge: disconnect() drains admitted leases before teardown, so a handler stuck
+ * forever previously made disconnect unreachable (only a force-stop recovered). The handler timeout must bound that
+ * wait.
+ */
+ @Test
+ fun `disconnect completes after a wedged handler is timed out`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+ val session = requireNotNull(service.activeSession.value)
+ val wedgeStarted = CompletableDeferred<Unit>()
+ val neverReleased = CompletableDeferred<Unit>()
+
+ val wedged = launch {
+ service.runWhileSessionActive(session) {
+ wedgeStarted.complete(Unit)
+ neverReleased.await()
+ }
+ }
+ wedgeStarted.await()
+
+ val disconnectJob = launch { service.disconnect() }
+ try {
+ testDispatcher.scheduler.runCurrent()
+ assertFalse(disconnectJob.isCompleted, "disconnect must wait while the lease is admitted")
+
+ // Cross the handler timeout (cancels the wedged block, draining the lease) plus the
+ // polite-disconnect drain window inside stopTransportLocked.
+ advanceTimeBy(121_000L)
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+ disconnectJob.join()
+ wedged.join()
+
+ assertNull(service.activeSession.value, "teardown must complete once the wedged lease is released")
+ } finally {
+ neverReleased.complete(Unit)
+ wedged.cancel()
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
@Test
fun `USB permission denial emits error and permanent disconnected state`() = runTest(testDispatcher) {
clock = 0L
Served by rngit 1.5.0 - Generated in 0.08s